package org.apache.rocketmq.proxy.service.message;

import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.lang3.NotImplementedException;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageBatch;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.common.ProxyException;
import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
import org.apache.rocketmq.proxy.service.channel.ChannelManager;
import org.apache.rocketmq.proxy.service.channel.InvocationContext;
import org.apache.rocketmq.proxy.service.channel.SimpleChannel;
import org.apache.rocketmq.proxy.service.route.AddressableMessageQueue;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.header.GetMaxOffsetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.SendMessageResponseHeader;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/17
 * @方法描述：本地消息服务组件，当proxy模块以本地模式部署时，就会使用这个类的对象把生产者客户端发送的消息专拣给部署在同一进程的Broker节点处理
 */
public class LocalMessageService implements MessageService {

    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);

    //Broker控制器，得到了这个控制器就相当于得到了Broker节点，可以把生产者生产的消息交给Broker节点处理
    //当然，在第十二版本代码我还没有实现这个功能
    private final BrokerController brokerController;

    //通道管理器，这个通道管理器管理的是Proxy模块和Broker节点之间数据传输的通道
    private final ChannelManager channelManager;


    public LocalMessageService(BrokerController brokerController, ChannelManager channelManager, RPCHook rpcHook) {
        this.brokerController = brokerController;
        this.channelManager = channelManager;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把生产者生产者消息转交给Broker节点处理的方法，在第十三版本代码中，该方法终于被实现了。
     */
    @Override
    public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, AddressableMessageQueue messageQueue,
                                                           List<Message> msgList, SendMessageRequestHeader requestHeader, long timeoutMillis) {
        //定义一个自己数组用来记录消息体
        byte[] body;
        //记录消息唯一标识，在回复客户端响应的时候会用到这个局部变量
        String messageId;
        //判断消息是否为批量消息
        if (msgList.size() > 1) {
            //如果是则设置批量消息标志，在第十三版本代码中，我们先不实现发送批量消息功能，只实现发送单条消息功能
            requestHeader.setBatch(true);
            //如果是批量消息，就在这里把存放消息的集合转换成存放批量消息的对象
            MessageBatch msgBatch = MessageBatch.generateFromList(msgList);
            //设置批量消息的唯一Id
            MessageClientIDSetter.setUniqID(msgBatch);
            //将批量消息编码为字节数组
            body = msgBatch.encode();
            //设置编码后的消息体
            msgBatch.setBody(body);
            //获取批量消息的唯一Id
            messageId = MessageClientIDSetter.getUniqID(msgBatch);
        } else {
            //执行到这里意味着要发送的是单条消息
            Message message = msgList.get(0);
            //得到消息体
            body = message.getBody();
            //得到消息的唯一Id
            messageId = MessageClientIDSetter.getUniqID(message);
        }
        //在这里把要发送的消息封装成RemotingCommand对象，因为消息最终是要发送给Broker节点的
        //而rocketmq内部各个模块通信使用的仍然是Remoting协议，所以要把消息封装成RemotingCommand对象发送给对方
        RemotingCommand request = LocalRemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader, ctx.getLanguage());
        //设置消息体
        request.setBody(body);
        //创建用来获得异步执行结果的future对象
        CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
        //在这里使用channel管理器创建了一个channel对象，这里我要重点解释一下
        //目前在proxy模块中已经有了一个grpcchannel管理器，其实还有一个nettychannel管理器，只不过我们并没有真的引入它
        //这两个channel管理器分别管理客户端与服务端grpc通信的channel和netty通信的channel
        //现在又多出来一个ChannelManager，而这个ChannelManager管理器就是用来管理Proxy模块和Broker模块之间数据传输的通道
        //大家一定要去分好这三个channel管理器的作用。这里调用了channelManager.createInvocationChannel(ctx)方法返回了一个SimpleChannel对象
        //实际上createInvocationChannel方法创建的是一个InvocationChannel对象，而InvocationChannel继承了SimpleChannel
        //这个InvocationChannel对象就是内部通道的意思，专门用来在proxy模块和broker模块内部通信，这里我还要再强调一下
        //只有当proxy以本地模式部署时，也就是proxy和broker部署在同一进程中，这个InvocationChannel才会发挥作用，这一点大家也要梳理清楚
        SimpleChannel channel = channelManager.createInvocationChannel(ctx);
        //创建InvocationChannel通道上下文对象，注意，这里把刚才创建的用来接收异步执行结果的future设置到了InvocationChannel通道上下文对象中
        //这个InvocationContext对象的handle方法会把Broker节点处理消息的响应设置给future对象，当future接收到响应后，就意味着proxy节点接收到了
        //borker模块返回的响应。所以，大家可以重点关注一下，InvocationContext对象的handle方法会在哪里被调用
        InvocationContext invocationContext = new InvocationContext(future);
        //把InvocationContext上下文对象注册到InvocationChannel内部的inFlightRequestMap成员变量中
        //key就是请求唯一Id，这样就可以根据响应唯一Id来判断proxy接收到的响应对应的是哪个请求的了
        channel.registerInvocationContext(request.getOpaque(), invocationContext);
        //得到channel处理上下文，这个处理上下文就是用来得到InvocationChannel对象的，等broker处理完消息后，会从这个上下文对象中得到InvocationChannel
        //然后调用InvocationChannel对象的writeAndFlush()方法将处理结果发回给proxy模块，大家可以先记住这个逻辑，后面就清楚是怎么回事了
        ChannelHandlerContext simpleChannelHandlerContext = channel.getChannelHandlerContext();
        try {
            //这里的操作就很明显了，就是直接得到broker的SendMessageProcessor消息处理器来处理生产者客户端发送的消息
            RemotingCommand response = brokerController.getSendMessageProcessor().processRequest(simpleChannelHandlerContext, request);
            //这里会返回空，因为消息存储到commitlog文件的操作是异步执行的
            if (response != null) {
                //这里可以清晰地看到，如果有响应，那就直接把响应设置到invocationContext上下文对象中
                invocationContext.handle(response);
                //清除channel中对应的上下文对象
                channel.eraseInvocationContext(request.getOpaque());
            }
        } catch (Exception e) {
            //发生异常则使用异常填充future对象
            future.completeExceptionally(e);
            channel.eraseInvocationContext(request.getOpaque());
            log.error("Failed to process sendMessage command", e);
        }
        //接下来就是把消息处理结果返回给生产者客户端的操作了，可以看到，这些操作被封装成了一个回调方法注册到刚才创建的future对象中了
        //这也就意味着只要proxy接收到了来自broker回复的消息处理结果，就会把结果封装成响应返回给生产者客户端
        //但在第十三版本代码中我们还不比关注响应如何回复给生产者客户端，因为现在broker处理消息后回复给proxy的结果并不完整
        //等到后面版本我们再真正实现回复生产者客户端响应的功能，所以下面的代码我就先不添加注释了，这些回复响应的代码也都不完整，后面还会重构
        //目前大家只需要关注消息是怎么被broker处理以及落盘即可
        return future.thenApply(r -> {
            SendResult sendResult = new SendResult();
            SendMessageResponseHeader responseHeader = (SendMessageResponseHeader) r.readCustomHeader();
            SendStatus sendStatus;
            switch (r.getCode()) {
                case ResponseCode.FLUSH_DISK_TIMEOUT: {
                    sendStatus = SendStatus.FLUSH_DISK_TIMEOUT;
                    break;
                }
                case ResponseCode.FLUSH_SLAVE_TIMEOUT: {
                    sendStatus = SendStatus.FLUSH_SLAVE_TIMEOUT;
                    break;
                }
                case ResponseCode.SLAVE_NOT_AVAILABLE: {
                    sendStatus = SendStatus.SLAVE_NOT_AVAILABLE;
                    break;
                }
                case ResponseCode.SUCCESS: {
                    sendStatus = SendStatus.SEND_OK;
                    break;
                }
                default: {
                    throw new ProxyException(ProxyExceptionCode.INTERNAL_SERVER_ERROR, r.getRemark());
                }
            }
            sendResult.setSendStatus(sendStatus);
            sendResult.setMsgId(messageId);
            sendResult.setMessageQueue(new MessageQueue(requestHeader.getTopic(), brokerController.getBrokerConfig().getBrokerName(), requestHeader.getQueueId()));
            //sendResult.setQueueOffset(responseHeader.getQueueOffset());
            //sendResult.setTransactionId(responseHeader.getTransactionId());
            //sendResult.setOffsetMsgId(responseHeader.getMsgId());
            return Collections.singletonList(sendResult);
        });
    }


    @Override
    public CompletableFuture<Long> getMaxOffset(ProxyContext ctx, AddressableMessageQueue messageQueue,
                                                GetMaxOffsetRequestHeader requestHeader, long timeoutMillis) {
        throw new NotImplementedException("getMaxOffset is not implemented in LocalMessageService");
    }

    @Override
    public CompletableFuture<RemotingCommand> request(ProxyContext ctx, String brokerName, RemotingCommand request,
                                                      long timeoutMillis) {
        throw new NotImplementedException("request is not implemented in LocalMessageService");
    }

    @Override
    public CompletableFuture<Void> requestOneway(ProxyContext ctx, String brokerName, RemotingCommand request,
                                                 long timeoutMillis) {
        throw new NotImplementedException("requestOneway is not implemented in LocalMessageService");
    }

}
